/**
 * Copyright 2016 vip.com.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
 * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations under the License.
 * </p>
 **/

package spring.cloud.tasks.api.service.helper;

import org.apache.curator.framework.CuratorFramework;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import spring.cloud.tasks.api.domain.NameSpaceClient;
import spring.cloud.tasks.api.domain.NameSpaceConfiguration;
import spring.cloud.tasks.api.service.ZooKeeperService;
import spring.cloud.tasks.api.utils.TaskNodePath;
import spring.cloud.tasks.api.utils.ThreadLocalCuratorClient;
import spring.cloud.tasks.api.zookeeper.ZooKeeper;
import spring.cloud.tasks.api.zookeeper.ZooKeeperRepository;

/**
 * @author hebelala
 */
public class ReuseUtils {

    private static final String JOB_NOT_EXIST_TEMPLATE = "The job {%s} does not exists.";

    private static final String NAMESPACE_NOT_EXIST_TEMPLATE = "The namespace {%s} does not exists.";

    private static final Logger log = LoggerFactory.getLogger(ReuseUtils.class);

    public static <T> T reuse(String namespace, final String jobName, ZooKeeperService zooKeeperService,
                              ZooKeeperRepository curatorRepository, final ZooKeeperReUseCallBack<T> callBack) throws RuntimeException {
        return reuse(namespace, zooKeeperService, curatorRepository, new ZooKeeperReUseCallBack<T>() {
            @Override
            public T call(ZooKeeper zooKeeper) throws RuntimeException {
                if (!zooKeeper.checkExists(TaskNodePath.getTaskNodePath(jobName))) {
                    throw new RuntimeException(String.format(JOB_NOT_EXIST_TEMPLATE, jobName));
                }
                return callBack.call(zooKeeper);
            }
        });
    }

    public static <T> T reuse(String namespace, ZooKeeperService zooKeeperService,
                              ZooKeeperRepository zooKeeperRepository, ZooKeeperReUseCallBack<T> callBack) throws RuntimeException {
        try {
            NameSpaceConfiguration registryCenterConfiguration = zooKeeperService
                    .getNameSpaceConfigurationByNameSpace(namespace);
            if (registryCenterConfiguration == null) {
                throw new RuntimeException(String.format(NAMESPACE_NOT_EXIST_TEMPLATE, namespace));
            }
            NameSpaceClient nameSpaceClient = zooKeeperService.connectByNameSpace(namespace);
            if (nameSpaceClient != null && nameSpaceClient.isConnected()) {
                CuratorFramework curatorFramework = nameSpaceClient.getCuratorFramework();
                ZooKeeper zooKeeper = zooKeeperRepository.getZooKeeper(curatorFramework);
                ThreadLocalCuratorClient.setCuratorFrameworkThreadLocal(curatorFramework);
                return callBack.call(zooKeeper);
            } else {
                throw new RuntimeException("Connect zookeeper failed");
            }
        } catch (RuntimeException e) {
            throw e;
        } catch (Throwable t) {
            log.error(t.getMessage(), t);
            throw new RuntimeException(t);
        } finally {
            ThreadLocalCuratorClient.clear();
        }
    }

    public static void reuse(String namespace, final String jobName, ZooKeeperService zooKeeperService,
                             ZooKeeperRepository curatorRepository, final ReuseCallBackWithoutReturn callBack)
            throws RuntimeException {
        reuse(namespace, zooKeeperService, curatorRepository, new ReuseCallBackWithoutReturn() {
            @Override
            public void call(ZooKeeper zooKeeper) throws RuntimeException {
                if (!zooKeeper.checkExists(TaskNodePath.getTaskNodePath(jobName))) {
                    throw new RuntimeException(String.format(JOB_NOT_EXIST_TEMPLATE, jobName));
                }
                callBack.call(zooKeeper);
            }
        });
    }

    public static void reuse(String namespace, ZooKeeperService zooKeeperService,
                             ZooKeeperRepository zooKeeperRepository, ReuseCallBackWithoutReturn callBack) throws RuntimeException {
        try {
            NameSpaceConfiguration registryCenterConfiguration = zooKeeperService
                    .getNameSpaceConfigurationByNameSpace(namespace);
            if (registryCenterConfiguration == null) {
                throw new RuntimeException(String.format(NAMESPACE_NOT_EXIST_TEMPLATE, namespace));
            }
            NameSpaceClient nameSpaceClient = zooKeeperService.connectByNameSpace(namespace);
            if (nameSpaceClient != null && nameSpaceClient.isConnected()) {
                CuratorFramework curatorFramework = nameSpaceClient.getCuratorFramework();
                ZooKeeper zooKeeper = zooKeeperRepository.getZooKeeper(curatorFramework);
                ThreadLocalCuratorClient.setCuratorFrameworkThreadLocal(curatorFramework);
                callBack.call(zooKeeper);
            } else {
                throw new RuntimeException("Connect zookeeper failed");
            }
        } catch (RuntimeException e) {
            throw e;
        } catch (Throwable t) {
            log.error(t.getMessage(), t);
            throw new RuntimeException(t);
        } finally {
            ThreadLocalCuratorClient.clear();
        }
    }

}
